package com.dk.common.distribute.election;

import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Slf4j
@RequiredArgsConstructor
public class RedissonLeaderElection implements DistLeaderElection {
    private final Set<ElectionListener> listeners = new HashSet<>();
    private final LeaderElection leaderElection;

    public RedissonLeaderElection(RLock leaderLock) {
        leaderElection = new LeaderElection(leaderLock, new ElectionListener() {
            @Override
            public void gotLeader() {
                for (ElectionListener listener : listeners) {
                    listener.gotLeader();
                }
            }

            @Override
            public void loseLeader() {
                for (ElectionListener listener : listeners) {
                    listener.loseLeader();
                }
            }
        });
        leaderElection.tryHold();
    }

    @Override
    public boolean isLeader() {
        return leaderElection.isMaster();
    }

    @Override
    public void addListener(ElectionListener electionListener) {
        if (!listeners.add(electionListener)) {
            return;
        }
        if (isLeader()) {
            electionListener.gotLeader();
        }
    }

    @Override
    public void removeListener(ElectionListener electionListener) {
        if (!listeners.remove(electionListener)) {
            return;
        }
        electionListener.loseLeader();
    }

    @SneakyThrows
    public void shutdown() {
        leaderElection.shutdown();
        for (ElectionListener listener : listeners) {
            listener.loseLeader();
        }
    }

    @Slf4j
    private static final class LeaderElection {
        private static final int WAIT_SECONDS = 1;
        private final Object masterLock = new Object();

        private final Object initLock = new Object();

        private final RLock leaderLock;

        private final ElectionListener listener;

        private final ElectionThread electionThread;

        private boolean stop = false;

        private boolean isInit = false;

        public LeaderElection(RLock leaderLock, ElectionListener listener) {
            this.leaderLock = leaderLock;
            this.listener = listener;
            this.electionThread = new ElectionThread();
        }


        void tryHold() {
            electionThread.start();
        }

        public boolean isMaster() {
            synchronized (initLock) {
                if (!isInit) {
                    try {
                        initLock.wait();
                    } catch (InterruptedException ignored) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
            return electionThread.isMaster();
        }

        public void shutdown() {
            if (stop) {
                return;
            }
            stop = true;
            try {
                synchronized (masterLock) {
                    masterLock.notifyAll();
                }
                electionThread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            log.info("shutdown and give up leadership");
        }

        private class ElectionThread extends Thread {
            private boolean isMaster = false;

            public ElectionThread() {
                setName(leaderLock.getName() + "-leader-election");
            }

            @Override
            public void run() {
                while (!stop) {
                    try {
                        if (isMaster) {
                            synchronized (masterLock) {
                                if (isInit) {
                                    masterLock.wait();
                                } else {
                                    masterLock.wait(Duration.ofSeconds(WAIT_SECONDS).toMillis());
                                }
                            }
                        } else {
                            isMaster = leaderLock.tryLock(WAIT_SECONDS, TimeUnit.SECONDS);
                            if (isMaster) {
                                log.debug("got leadership");
                                listener.gotLeader();
                            }
                        }
                    } catch (InterruptedException ignored) {
                        Thread.currentThread().interrupt();
                    } finally {
                        synchronized (initLock) {
                            if (!isInit) {
                                initLock.notifyAll();
                                isInit = true;
                            }
                        }
                    }
                }

                if (leaderLock.isLocked() && leaderLock.isHeldByCurrentThread()) {
                    leaderLock.unlock();
                }

                if (isMaster) {
                    isMaster = false;
                }
            }

            public boolean isMaster() {
                return isMaster;
            }
        }
    }
}
